package com.pai4j.search.service.es;


import com.google.common.collect.Lists;
import com.pai4j.common.bean.PAIPageResponseBeanUtil;
import com.pai4j.common.util.JsonUtil;
import com.pai4j.domain.vo.dto.AbstractBaseEsDTO;
import com.pai4j.domain.vo.dto.AbstractESMessageDTO;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.MultiMatchQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.TermQueryBuilder;
import org.elasticsearch.index.query.Operator;
import org.elasticsearch.common.unit.Fuzziness;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightField;
import org.elasticsearch.search.sort.SortOrder;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.action.support.WriteRequest;
import org.springframework.beans.factory.annotation.Value;

import java.io.IOException;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 *  抽象ES操作服务，不同搜索业务场景自己继承实现
 */
@Slf4j
public abstract class AbstractPai4jElasticSearchClient<T extends AbstractBaseEsDTO> {

    @Resource
    private RestHighLevelClient client;

    @Value("${spring.elasticsearch.index.shards:1}")
    private Integer cfgShards;
    @Value("${spring.elasticsearch.index.replicas:1}")
    private Integer cfgReplicas;

   
    /**
     * 批量保存操作的最大尺寸限制
     * 此常量用于控制批量保存操作时，单次保存的数据量大小
     * 保持较小的批量大小可以减少内存消耗和提高数据库插入效率
     */
    private static final int BULK_SAVE_SIZE = 10;

    /**
     * 实现类中配置不同的索引
     *
     * @return
     */
    protected abstract String getIndexName();

    /**
     * 搜索json结果反序列化为T类型对象
     *
     * @param dataJson
     * @return
     */
    protected abstract T convert(String dataJson);

    /**
     * 填充扩展信息
     *
     * @param dataList
     */
    protected abstract void setSearchResultExtInfo(List<T> dataList);

    /**
     * 获取分片数量
     *
     * 此方法用于配置或查询系统中的分片数量
     * 分片是将数据分布在多个节点上的技术，可以提高系统的可扩展性和性能
     *
     * @return 分片的数量，默认返回1，表示没有进行数据分片
     */
    protected Integer getShards() {
        return cfgShards == null ? 1 : cfgShards;
    }
    
    /**
     * 获取副本数量
     *
     * 此方法用于配置或查询系统中每个分片的副本数量
     * 副本用于提高系统的容错性，通过复制数据到多个节点，确保在部分节点故障时数据仍然可用
     *
     * @return 每个分片的副本数量，默认返回1，表示没有设置副本
     */
    protected Integer getReplicas() {
        return cfgReplicas == null ? 1 : cfgReplicas;
    }

    /**
     * Bulk 刷新策略，默认 NONE，可由子类覆盖
     */
    protected WriteRequest.RefreshPolicy getRefreshPolicy() {
        return WriteRequest.RefreshPolicy.NONE;
    }

    /**
     * 获取索引的设置信息
     *
     * 此方法用于构建和返回一个包含索引设置的Settings对象
     * 它主要配置了索引的分片数和副本数，这两个参数是Elasticsearch索引配置中非常重要的属性
     *
     * @return Settings对象，包含了索引的分片数和副本数配置
     */
    protected Settings getSettings() {
        return Settings.builder()
                .put("number_of_shards", this.getShards())
                .put("number_of_replicas", this.getReplicas())
                .build();
    }

    /**
     * 具体搜索服务自己实现
     * @return
     */
    protected abstract XContentBuilder getXContentBuilder() throws IOException;


    /**
     * ES 搜索
     *
     * @param type   搜索数据类型：聊天消息/用户/公众号文章/群组
     * @param userId 搜索用户
     * @param q      搜索关键词
     * @param page   搜索页码
     * @param size   搜索条数
     * @return
     */
    public PAIPageResponseBeanUtil<T> query(Integer type, String userId, String q , int page, int size) {
        SearchRequest searchRequest = new SearchRequest(getIndexName());
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        // 创建HighlightBuilder并设置高亮标签
        HighlightBuilder highlightBuilder = new HighlightBuilder();
        // 高亮开始标签
        highlightBuilder.preTags("<span style='color:red'>");
        // 高亮结束标签
        highlightBuilder.postTags("</span>");
        // 明确高亮字段
        highlightBuilder.field("content");

        // 创建multi-match查询
        MultiMatchQueryBuilder multiMatchQuery =
                QueryBuilders.multiMatchQuery(q, "content")
                        .analyzer("ik_smart") //设置分析器为ik_smart
                        .operator(Operator.AND)
                        .minimumShouldMatch("75%")
                        .fuzziness(Fuzziness.AUTO);

        // 创建level的term查询
        TermQueryBuilder senderIdQuery = QueryBuilders.termQuery("sender_id", userId);
        TermQueryBuilder receiverIdQuery = QueryBuilders.termQuery("receiver_id", userId);
        TermQueryBuilder proxySenderIdQuery = QueryBuilders.termQuery("proxy_sender_id", userId);

        // 将两个查询合并为一个bool查询，使用should并设置minimumShouldMatch为1
        BoolQueryBuilder userIdBoolQuery = QueryBuilders.boolQuery()
                .should(senderIdQuery)
                .should(receiverIdQuery)
                .should(proxySenderIdQuery)
                .minimumShouldMatch(1);

        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery()
                .must(multiMatchQuery);
        // userId 判空处理
        if (StringUtils.isNotBlank(userId)) {
            boolQueryBuilder.must(userIdBoolQuery);
            // 可选routing提高性能
            searchRequest.routing(userId);
        }
        // type 过滤
        if (type != null) {
            boolQueryBuilder.must(QueryBuilders.termQuery("type", type));
        }

        // 应用查询
        searchSourceBuilder.query(boolQueryBuilder);
        searchSourceBuilder.highlighter(highlightBuilder);
        // 设置分页参数
        // 计算偏移量
        searchSourceBuilder.from((page - 1) * size);
        searchSourceBuilder.size(size);
        // 按照相关性排序
        searchSourceBuilder.sort("_score", SortOrder.DESC);
        // 性能与稳定性
        searchSourceBuilder.trackTotalHitsUpTo(10_000);
        searchSourceBuilder.timeout(TimeValue.timeValueSeconds(5));
        searchRequest.source(searchSourceBuilder);
        // 执行搜索
        List<T> results = new ArrayList<>();
        long total = 0L;
        try {
            SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
            total = searchResponse.getHits().getTotalHits().value;
            if (total > 0) {
                for (SearchHit hit : searchResponse.getHits()) {
                    Map<String, HighlightField> highlightFields = hit.getHighlightFields();
                    Map<String, Object> searchRespMap = hit.getSourceAsMap();
                    String dataJson = String.valueOf(searchRespMap.get("data"));
                    T dto = this.convert(dataJson);
                    // 返回高亮结果
                    HighlightField hf = highlightFields == null ? null : highlightFields.get("content");
                    if (hf != null && hf.fragments() != null && hf.fragments().length > 0) {
                        try {
                            // 假设DTO有message字段作为显示内容
                            Field msgField = dto.getClass().getDeclaredField("message");
                            msgField.setAccessible(true);
                            msgField.set(dto, hf.fragments()[0].string());
                        } catch (Exception ignore) {
                            // 反射失败不影响主流程
                        }
                    }
                    results.add(dto);
                }
            }
        } catch (IOException e) {
            // 版本兼容问题，不影响结果
        }
        // 填充扩展信息
        this.setSearchResultExtInfo(results);
        return PAIPageResponseBeanUtil.success(page, size, total, results);
    }

    /**
     * 增量数据同步ES
     *
     */
    public void sync(AbstractESMessageDTO dto) {
        
        this.saveDocument(dto);
    }

    /**
     * 批量同步数据到ES，常被用于增量数据同步场景
     * @param dtoList
     */
    public void bulkSync(List<AbstractESMessageDTO> dtoList) {
        
        this.bulkSaveDocument(dtoList);
    }

    /**
     * 创建 ES 索引
     *
     * 本方法负责在 Elasticsearch 中创建一个指定名称和映射的索引
     * 它首先获取索引名称和映射内容，然后将这些信息封装到创建索引请求中，
     * 最后通过客户端发送请求并返回创建结果
     *
     * @return boolean 表示索引是否创建成功
     */
    public boolean createIndex() {
        // 获取索引名称
        String indexName = this.getIndexName();
        // 初始化映射内容
        XContentBuilder mapping = null;
        try {
            mapping = this.getXContentBuilder();
        } catch (IOException e) {
            // 如果映射内容构建失败，则记录错误日志并抛出运行时异常
            log.error("ES索引创建失败 =====> indexName:{}", indexName, e);
            throw new RuntimeException(e);
        }
        // 存在性检查
        try {
            boolean exists = client.indices().exists(new GetIndexRequest(indexName), RequestOptions.DEFAULT);
            if (exists) {
                log.info("ES索引已存在，跳过创建 =====> indexName:{}", indexName);
                return true;
            }
        } catch (IOException e) {
            log.warn("检查索引存在性失败，将继续尝试创建 =====> indexName:{}", indexName, e);
        }
        // 创建索引请求，包含索引名称、设置和映射
        CreateIndexRequest createIndexRequest = new CreateIndexRequest(indexName)
                                                .settings(this.getSettings())
                                                .mapping(mapping.toString());
        try {
            // 发送创建索引请求并返回结果
            return client.indices().create(createIndexRequest, RequestOptions.DEFAULT).isAcknowledged();
        } catch (Exception e) {
            // 如果索引创建失败，则记录错误日志并抛出运行时异常
            log.error("ES索引创建失败 =====> indexName:{}", indexName, e);
            throw new RuntimeException(e);
        }
    }
    /**
     * 删除索引
     *
     * 本方法旨在删除Elasticsearch中的一个指定索引它首先获取索引名称，
     * 然后构造一个删除索引的请求，最后执行该请求并返回删除操作是否被确认
     *
     * @return boolean 返回删除操作是否被确认如果返回true，表示索引删除成功；
     *         如果抛出RuntimeException异常，表示索引删除过程中出现错误
     */
    public boolean deleteIndex() {
        // 获取当前实例的索引名称
        String indexName = this.getIndexName();
        // 不存在则幂等返回
        try {
            boolean exists = client.indices().exists(new GetIndexRequest(indexName), RequestOptions.DEFAULT);
            if (!exists) {
                log.info("ES索引不存在，跳过删除 =====> indexName:{}", indexName);
                return true;
            }
        } catch (IOException e) {
            log.warn("检查索引存在性失败，仍尝试删除 =====> indexName:{}", indexName, e);
        }
        // 创建删除索引请求对象，指定要删除的索引名称
        DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest(indexName);
        try {
            // 执行删除索引操作，并返回操作是否被确认
            return client.indices().delete(deleteIndexRequest, RequestOptions.DEFAULT).isAcknowledged();
        } catch (IOException e) {
            // 记录索引删除失败的日志信息
            log.error("ES索引删除失败 =====> indexName:{}", indexName, e);
            // 将捕获到的异常重新抛出，以便调用者能够处理
            throw new RuntimeException(e);
        }
    }
    /**
     * （单条记录写入）保存文档数据到ES
     *
     * 该方法用于将单个文档对象的数据保存到Elasticsearch中它接受一个抽象的ES消息DTO对象作为参数，
     * 根据DTO对象中的数据类型和业务ID生成文档ID，并将DTO对象转换为JSON格式的数据，然后通过ES客户端
     * 的index方法将数据保存到指定的索引中
     *
     * @param dto ES消息的数据传输对象，包含了要保存到ES的文档数据
     */
    public void saveDocument(AbstractESMessageDTO dto) {
        // 生成文档ID，格式为"type_bizId"，以确保ID的唯一性
        String documentId = String.format("%s_%s", dto.getType(), dto.getBizId());
        
        // 创建索引请求对象，指定索引名、文档类型、文档ID，并设置文档的JSON格式数据
        IndexRequest indexRequest = new IndexRequest(this.getIndexName(), "_doc", documentId).id(documentId);
        indexRequest.source(JsonUtil.toJsonString(dto), XContentType.JSON);
        
        // 执行索引操作，将文档数据保存到ES中
        try {
            client.index(indexRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            // 异常处理：如果发生IOException，此处应记录日志或进行其他处理
        }
    }

    /**
     * （批量）文档数据批量写入ES
     *
     * @param dtoList 一个包含多个AbstractESMessageDTO对象的列表，用于批量保存到ES
     */
    public void bulkSaveDocument(List<AbstractESMessageDTO> dtoList) {
        // 将dtoList列表按照BULK_SAVE_SIZE大小进行分块处理
        for (List<AbstractESMessageDTO> part : Lists.partition(dtoList, BULK_SAVE_SIZE)) {
            BulkRequest bulkRequest = new BulkRequest();
            bulkRequest.setRefreshPolicy(this.getRefreshPolicy());
            for (AbstractESMessageDTO data : part) {
                String documentId = String.format("%s_%s", data.getType(), data.getBizId());
                IndexRequest indexRequest = new IndexRequest(this.getIndexName())
                        .id(documentId)
                        .source(JsonUtil.toJsonString(data), XContentType.JSON);
                bulkRequest.add(indexRequest);
            }
            try {
                client.bulk(bulkRequest, RequestOptions.DEFAULT);
            } catch (IOException e) {
                log.error("ES 文档数据批量写入ES异常 =====> indexName:{}, partSize:{}, part:{}",
                        this.getIndexName(), part.size(), JsonUtil.toJsonString(part), e);
                throw new RuntimeException(e);
            }
        }
    }

    /**
     * 按ID删除文档
     */
    public void deleteById(Integer type, String bizId) {
        String documentId = String.format("%s_%s", type, bizId);
        DeleteRequest request = new DeleteRequest(this.getIndexName(), documentId);
        try {
            client.delete(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            log.error("ES 文档删除异常 =====> indexName:{}, id:{}", this.getIndexName(), documentId, e);
            throw new RuntimeException(e);
        }
    }

    /**
     * 按ID更新文档（局部更新，若不存在则插入）
     */
    public void updateById(AbstractESMessageDTO dto) {
        String documentId = String.format("%s_%s", dto.getType(), dto.getBizId());
        UpdateRequest request = new UpdateRequest(this.getIndexName(), documentId)
                .doc(JsonUtil.toJsonString(dto), XContentType.JSON)
                .docAsUpsert(true)
                .setRefreshPolicy(this.getRefreshPolicy());
        try {
            client.update(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            log.error("ES 文档更新异常 =====> indexName:{}, id:{}", this.getIndexName(), documentId, e);
            throw new RuntimeException(e);
        }
    }



}
